package edu.xidian.sei.mlcs

import java.io.File
import java.util.concurrent.ForkJoinPool
import java.util.concurrent.ForkJoinTask
import java.util.concurrent.RecursiveTask

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.HashMap
import scala.collection.mutable.ListBuffer
import scala.collection.mutable.Queue
import scala.concurrent.Await
import scala.concurrent.duration.DurationInt

import com.typesafe.config.ConfigFactory
import akka.pattern.ask
import akka.util.Timeout

import scala.concurrent.Await
import scala.concurrent.duration.DurationInt
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.util.Timeout
import edu.xidian.sei.mlcs.util.Stopwatch

object LocalClient {

  val actorNum = 5
  val actorList = new ArrayBuffer[SubGraphActor]

  def main(args: Array[String]): Unit = {
    if (args.length == 0) {
      println(
        "Usage:HorizontalCrawler /path/to/your/data/file -Dmlcs.max-thread=2 -Dmlcs.auto-shrink=true -Dmlcs.optimal-only=false");
      return ;
    }

    val prompts = new StringBuilder

    val mlcs = MLCS.loadFromFile(args(0))
    mlcs.maxThread = 2

    val maxThread = System.getProperty("mlcs.max-thread")
    if (null != maxThread) {
      mlcs.maxThread = Integer.parseInt(maxThread);
      prompts.append(" max-thread=" + maxThread);
    }

    println("Starting search " + args(0) + prompts)

    implicit val timeout = Timeout(10 seconds)

    val startTime = System.nanoTime()
    val system = ActorSystem.apply("LocalClient")
    val addressMap = new HashMap[Int, ActorRef]
    val config = ConfigFactory.parseFile(new File("subgraph.conf")).getConfig("subgraphactor")
    val addresses = config.getString("params.addresses").split("|")
    addresses.foreach { add =>
      val strs = add.split("@")
      val otherActor = Await.result(system.actorSelection(s"akka.tcp://SubGraph@${strs(1)}/user/subgraph").resolveOne(), timeout.duration)
      addressMap.put(strs(0).toInt, otherActor)
    }

    addressMap(mlcs.start.indexSum % addressMap.size) ! InitialNode(mlcs.start)
    addressMap(mlcs.end.indexSum % addressMap.size) ! InitialNode(mlcs.end)
    buildGraph(mlcs, addressMap, mlcs.start)

    mlcs.nextSortedLocations(mlcs.start).foreach { loc =>
      addressMap(loc.indexSum % addressMap.size) ! ForwardTopoOrdering(mlcs.start, loc, 1)
    }

    var state = true
    while (state) {
      Thread.sleep(10000)
      val state = Await.result((addressMap(mlcs.end.indexSum % addressMap.size) ? ForwardTopoFinish).mapTo[Boolean], timeout.duration)
    }

    val result = findKeyLocs(mlcs, addressMap)
    System.out.println(result._1 + " mlcs(max length " + result._2 + ") found, using "
      + Stopwatch.format(System.nanoTime() - startTime));
    addressMap.values.foreach { v =>
      v ! "finish"
    }
  }

  def buildGraph(mlcs: MLCS, addressMap: HashMap[Int, ActorRef], head: Location) {
    implicit val searchTimeout = Timeout(5 seconds)
    val pool = if (mlcs.maxThread > 0) new ForkJoinPool(mlcs.maxThread) else new ForkJoinPool()
    val levelQueue = new ListBuffer[Location]()
    levelQueue += head
    val parallelism = pool.getParallelism();
    val startTime = System.nanoTime()
    while (!levelQueue.isEmpty) {
      val countPerSlice = Math.ceil(levelQueue.size * 1.0 / parallelism).toInt
      val locsList = levelQueue.sliding(countPerSlice, countPerSlice)
      val tasks = new ListBuffer[ForkJoinTask[ListBuffer[Location]]]
      locsList.foreach { locs =>
        tasks += pool.submit(new BFSCrawler(mlcs, locs))
      }
      levelQueue.clear
      tasks.foreach { task =>
        val nextLocs = task.join()
        nextLocs.foreach { loc =>
          val state = Await.result((addressMap(loc.indexSum % addressMap.size) ? BuildGraph(loc)).mapTo[Boolean], searchTimeout.duration)
          if (state) levelQueue += loc
        }
      }

    }
    pool.shutdown
    println("build graph using " + Stopwatch.format(System.nanoTime() - startTime))
  }

  def findKeyLocs(mlcs: MLCS, addressMap: HashMap[Int, ActorRef]): Tuple2[BigDecimal, Int] = {
    implicit val searchTimeout = Timeout(5 seconds)
    val queue = new Queue[Tuple3[ListBuffer[Location], Location, BigDecimal]]

    var tlevel = 0
    // 从虚拟终点到该节点的可选路径数，初始终点为1

    var queueEnd = mlcs.end
    val tuple = Await.result((addressMap(mlcs.end.indexSum % addressMap.size) ? ReverseTopoENDNode).mapTo[Tuple2[ListBuffer[Location], Int]], searchTimeout.duration)
    queue.enqueue(Tuple3(tuple._1, mlcs.end, BigDecimal(1)))
    val maxLevel = tuple._2 - 1
    while (!queue.isEmpty) {
      val tuple3 = queue.dequeue()
      tuple3._1.foreach { ploc =>
        val tempTuple = Await.result((addressMap(ploc.indexSum % addressMap.size) ? ReverseTopoOrdering(ploc, tlevel, maxLevel, tuple3._3)).mapTo[Tuple3[ListBuffer[Location], Location, BigDecimal]], searchTimeout.duration)
        if (!tempTuple._1.isEmpty) {
          queue.enqueue(tempTuple)
        }

      }

      if (tuple3._2 equals (queueEnd)) {
        if (!queue.isEmpty) queueEnd = queue.last._2
        tlevel += 1
      }
    }
    val count = Await.result((addressMap(mlcs.start.indexSum % addressMap.size) ? ReverseTopoFinish).mapTo[BigDecimal], searchTimeout.duration)
    (count, maxLevel)

  }

}

class BFSCrawler(mlcs: MLCS, locations: ListBuffer[Location]) extends RecursiveTask[ListBuffer[Location]] {
  def compute(): ListBuffer[Location] = {
    val nexts = new ListBuffer[Location] //下次拓扑排序的点的层和点集
    locations.foreach { loc => //当前点的loc
      nexts ++= mlcs.nextLocations(loc)
    }
    nexts //返回一个层和改层点的映射
  }
}